package com.qch.kafka.demo;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.*;
import java.util.function.BiConsumer;

/**
 * @Author qch
 * @Date 2021/5/31
 */
public class MyOffsetManager {
    private static String topicName="tp_demo_01";
    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<String, Object>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.142.135:9092");
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, "mygrp1");
        configs.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer_id");
        //configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(configs);
       // consumer.subscribe(Collections.singleton("tp_demo_01"));
        //手动给消费者分配分区
        //1.需要知道有哪些主题可以访问和消费
        /*Map<String, List<PartitionInfo>> stringListMap = consumer.listTopics();
        stringListMap.forEach(new BiConsumer<String, List<PartitionInfo>>() {
            @Override
            public void accept(String topicName, List<PartitionInfo> partitionInfos) {
                System.out.println("主题名称"+topicName);
                for (PartitionInfo partitionInfo : partitionInfos) {
                    System.out.println(partitionInfo);
                }
            }
        });*/
     /*   Set<TopicPartition> assignment = consumer.assignment();
        for (TopicPartition topicPartition : assignment) {
            System.out.println(topicPartition);
        }
        System.out.println("--------------------------------------------");*/
        consumer.assign(Arrays.asList(
                new TopicPartition("tp_demo_01",0),
                new TopicPartition("tp_demo_01",2)
        ));
     /*   Set<TopicPartition> assignment2 = consumer.assignment();
        for (TopicPartition topicPartition : assignment2) {                new TopicPartition("tp_demo_01",1),

            System.out.println(topicPartition);
        }*/
        /**
         * 查看当前消费者在指定主题的分区上的消费者偏移量
         */
        long offset0 = consumer.position(new TopicPartition(topicName, 0));
        System.out.println("分区0的偏移量"+offset0);
        long offset1 = consumer.position(new TopicPartition(topicName, 1));
        System.out.println("分区1的偏移量"+offset1);
        /*consumer.seekToBeginning(Arrays.asList(new TopicPartition(topicName,0),
                new TopicPartition(topicName,2)
        ));*/
        /*consumer.seekToEnd(Arrays.asList(new TopicPartition(topicName,2)
        ));*/
        long offset2 = consumer.position(new TopicPartition(topicName, 2));
        System.out.println("分区2的偏移量"+offset2);
        consumer.close();
    }
}
